package pipe;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.Pipe;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.locks.ReentrantLock;

public class PipeClient {
    public static void main1(String[] args) {
        try {

            Pipe queue1 = Pipe.open();
            Pipe queue2 = Pipe.open();
            Selector selector = Selector.open();
            queue1.source().configureBlocking(false);
            queue1.sink().configureBlocking(false);
            queue1.source().register(selector, SelectionKey.OP_READ);
            queue2.source().configureBlocking(false);
            queue2.sink().configureBlocking(false);
            queue2.source().register(selector, SelectionKey.OP_READ);
            new Thread(() -> {
                int round = 2;
                SelectionKey key;
                while (!Thread.currentThread().isInterrupted()) {
                    try {
                        int select = selector.select();
                        if (select == 0) continue;
                        Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                        ByteBuffer buffer = ByteBuffer.allocate(16);
                        while (iterator.hasNext()) {
                            key = iterator.next();
                            Pipe.SourceChannel channel =(Pipe.SourceChannel) key.channel();
                            System.out.println(Thread.currentThread().getName()+" : "+channel.validOps());
                            for (int i = 0; i < round; i++) {
                                int cl = channel.read(buffer);
                                if (cl==0) break;
                                if (cl==-1) {
                                    key.cancel();
                                    break;
                                }
                                buffer.flip();
                                byte[] bytes = new byte[buffer.remaining()];
                                buffer.get(bytes);
                                System.out.println(Thread.currentThread().getName()+" :" +new String(bytes));

                                buffer.clear();
                            }

                            iterator.remove();
                        }

                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }


            }, "reader").start();
            new Thread(()->{
                Pipe.SinkChannel sink = queue1.sink();
                while (true){
                    String k = "msg_from_"+Thread.currentThread().getName()+System.currentTimeMillis();
                    ByteBuffer buffer = ByteBuffer.wrap(k.getBytes());
                    while (buffer.hasRemaining()){
                        System.out.println(Thread.currentThread().getName()+" : "+k);
                        try {
                            sink.write(buffer);
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                    try {
                        Thread.sleep(3000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }


                }

            },"writer1").start();
            new Thread(()->{
                Pipe.SinkChannel sink = queue2.sink();
                while (true){
                    String k = "msg_from_"+Thread.currentThread().getName()+System.currentTimeMillis();
                    ByteBuffer buffer = ByteBuffer.wrap(k.getBytes());
                    while (buffer.hasRemaining()){
                        System.out.println(Thread.currentThread().getName()+" : "+k);
                        try {
                            sink.write(buffer);
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                    try {
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }


                }

            },"writer2").start();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws IOException {
        Selector selector = Selector.open();
        new Thread(new Consumer(selector),"consumer").start();
        List<Pipe> pipes = new ArrayList<>();
        for (int i = 0; i < 3; i++) {
            Pipe pipe = Pipe.open();
            pipe.source().configureBlocking(false);
            pipe.sink().configureBlocking(false);
            pipe.source().register(selector,SelectionKey.OP_READ);
            selector.wakeup();
            pipes.add(pipe);
        }
        new Thread(new Producer(pipes.get(0),1000),"producer0").start();
        new Thread(new Producer(pipes.get(1),2000),"producer1").start();
        new Thread(new Producer(pipes.get(2),3000),"producer2").start();

    }
}

